package com.atguigu.dga.assess.assess.cal;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.dga.assess.assess.AssessorTemplate;
import com.atguigu.dga.assess.bean.AssessParam;
import com.atguigu.dga.assess.bean.GovernanceAssessDetail;
import com.atguigu.dga.config.MetaDataUtil;
import com.atguigu.dga.config.SqlParser;
import com.atguigu.dga.ds.bean.TDsTaskDefinition;
import com.atguigu.dga.meta.bean.TableMetaInfo;
import com.google.common.collect.Sets;
import lombok.Data;
import org.apache.hadoop.hive.ql.lib.Dispatcher;
import org.apache.hadoop.hive.ql.lib.Node;
import org.apache.hadoop.hive.ql.parse.ASTNode;
import org.apache.hadoop.hive.ql.parse.HiveParser;
import org.apache.hadoop.hive.ql.parse.SemanticException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.*;

@Component("SIMPLE_PROCESS")
public class CheckSimleProcess extends AssessorTemplate
{
    @Autowired
    private MetaDataUtil metaDataUtil;
    /*
        1.获取到当前表所用到的处理的sql。
        2.解析sql
            遍历sql的每一个节点。
                a)收集： group by,union, join等操作
                b)收集: where过滤的字段名
                            要和查询的表的分区进行比对，只要包含非分区字段，就是复杂查询，不用打0分
                c)收集： 这条sql查询的源表名
        3.对收集的结果进行处理
            如果有复杂操作的运算符，不管。
                否则，判断where中是否有非分区字段
                    没有，打0分。
     */
    @Override
    protected void assess(AssessParam param, GovernanceAssessDetail detail) throws Exception {

        TableMetaInfo tableMetaInfo = param.getTableMetaInfo();
        String schemaName = tableMetaInfo.getSchemaName();
        //获取当前表的定义
        TDsTaskDefinition taskDefinition = metaDataUtil.taskDefinitionMap.get(schemaName + "." + tableMetaInfo.getTableName());

        //load 的表都没有任务定义。 ods层的表使用一个脚本，装载所有
        if (taskDefinition == null){
            return ;
        }

        //1.获取到当前表所用到的处理的sql
        String sql = taskDefinition.getSql();

        //解析
        MyDispatcher myDispatcher = new MyDispatcher();
        SqlParser.parseSql(sql,myDispatcher);
        //获取解析的结果
        Set<String> complexOperators = myDispatcher.getComplexOperators();
        Set<String> whereFieldNames = myDispatcher.getWhereFieldNames();
        Set<String> fromTableNames = myDispatcher.getFromTableNames();

        detail.setAssessComment("当前sql含有以下复杂运算: "+complexOperators +",where过滤的字段: "+whereFieldNames+", 从以下表进行了查询:"+fromTableNames);

        //判断是否是简单查询
        if (complexOperators.isEmpty()){
            //没有复杂处理，判断where的字段中是否有非分区字段
            //获取所查询表的分区字段
            Set<String> partitions = new HashSet<>();
            fromTableNames
                    .stream()
                    .forEach(tableName -> {
                        String partitionColNameJson = metaDataUtil.tableMetaInfoMap.get(schemaName + "." + tableName).getPartitionColNameJson();
                        // [ {"name":"dt","comment":""},{} ]
                        List<JSONObject> jsonObjects = JSON.parseArray(partitionColNameJson, JSONObject.class);
                        for (JSONObject jsonObject : jsonObjects) {
                            partitions.add(jsonObject.getString("name"));
                        }
                    });

            /*
                和where过滤的字段进行比对。做差集运算。
                    whereSet: [a,b,c]
                    partitionsSet: [a,b ]

                     [a,b,c] 差集  [a,b ] = [c]

                     [a,b] 差集  [a,b ] = [ ]

                     set集合交集: retainAll()
                     set集合差集: removeAll()
             */
            whereFieldNames.removeAll(partitions);

            if (whereFieldNames.isEmpty()){
                //除了分区字段，没有再用过其他非分区字段过滤
                assessZeroScore(detail,"当前是简单查询!",param,false);
            }
        }

    }

    //创建每个节点的处理逻辑
    @Data
    public static class MyDispatcher implements Dispatcher{

        //创建集合收集资源
        //a)收集复杂操作
        private Set<String> complexOperators = new HashSet<>();

        //a)使用这个集合判断当前节点是不是复杂查询
        Set<Integer> realProcessSet= Sets.newHashSet(
                HiveParser.TOK_JOIN,  //join 包含通过where 连接的情况
                HiveParser.TOK_GROUPBY,       //  group by
                HiveParser.TOK_LEFTOUTERJOIN,       //  left join
                HiveParser.TOK_RIGHTOUTERJOIN,     //   right join
                HiveParser.TOK_FULLOUTERJOIN,     // full join
                HiveParser.TOK_FUNCTION,     //count(1)
                HiveParser.TOK_FUNCTIONDI,  //count(distinct xx)
                HiveParser.TOK_FUNCTIONSTAR, // count(*)
                HiveParser.TOK_SELECTDI,  // distinct
                HiveParser.TOK_UNIONALL   // union
        );

        //b)收集: where过滤的字段名
        private Set<String> whereFieldNames = new HashSet<>();
        //b) 判断当前节点是不是 比较运算符 =,>=,<=
        Set<String> operators= Sets.newHashSet("=",">","<",">=","<=" ,"<>"  ,"like","not like"); // in / not in 属于函数计算

        //c)收集： 这条sql查询的源表名
        private Set<String> fromTableNames = new HashSet<>();



        /*
            对每一个节点都执行操作
         */
        @Override
        public Object dispatch(Node nd, Stack<Node> stack, Object... nodeOutputs) throws SemanticException {

            ASTNode astNode = (ASTNode) nd;

            //判断当前是不是复杂查询的节点
            if (realProcessSet.contains(astNode.getType())){
                complexOperators.add(astNode.getText());
            }

            //判断是不是where节点，如果是尝试获取where节点的字段
            if (astNode.getType() == HiveParser.TOK_WHERE){
                getWhereFields(astNode);
            }

            /*
                判断当前节点是不是一个表引用节点
                    student
                    student t1
                    gmall.student t1
             */
            if (astNode.getType() == HiveParser.TOK_TABNAME){
                //没库名
                if (astNode.getChildren().size() == 1){
                    ASTNode child = (ASTNode) astNode.getChild(0);
                    fromTableNames.add(child.getText());
                }else {
                    //有库名
                    ASTNode child = (ASTNode) astNode.getChild(1);
                    fromTableNames.add(child.getText());
                }

            }
            return null;
        }

        private void getWhereFields(ASTNode astNode) {

            //获取where下面的一层
            ArrayList<Node> children = astNode.getChildren();

            //进行空值判断，当前这个节点可能是最后一层，没有孩子
            if (children == null ){
                return;
            }

            //这层可能是and | or之类的逻辑运算符
            for (Node child : children) {
                ASTNode childAstNode = (ASTNode) child;
                //判断是不是想要的 比较运算符
                if (operators.contains(childAstNode.getText())){
                    //当前是比较运算符，向下获取所有的子节点。
                    ArrayList<Node> operatorChildNodes = childAstNode.getChildren();
                    for (Node node : operatorChildNodes) {

                        ASTNode operatorChildAstNode = (ASTNode) node;
                        /*
                               where t1.age = 20
                                       =
                                      /\
                                     .  20
                                    /\
                      TOK_TABLE_OR_COL age
                                  /
                                 t1

                         */
                        if (operatorChildAstNode.getType() == HiveParser.DOT){
                            whereFieldNames.add(operatorChildAstNode.getChild(1).getText());
                        }

                        /*
                               where age = 20
                                       =
                                      /\
                      TOK_TABLE_OR_COL  20
                                     /
                                    age

                         */
                        else if (operatorChildAstNode.getType() == HiveParser.TOK_TABLE_OR_COL){
                            whereFieldNames.add(operatorChildAstNode.getChild(0).getText());
                        }
                    }
                }else {
                    //当前不是比较运算符，可能是逻辑运算符 and,or之类，再递归判断
                    getWhereFields(childAstNode);
                }
            }
        }
    }
}

























